package com.foreveross.fetl.queue;

import java.io.File;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.log4j.Logger;
import org.apache.tools.zip.ZipOutputStream;

import com.foreveross.fetl.common.cfg.system.ReportProperties;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.FetlErrorMessage;
import com.foreveross.fetl.common.vo.hangxin.PlaneInfo;
import com.foreveross.fetl.common.vo.model.doub.DoublePlaneInfo;
import com.foreveross.fetl.queue.util.ExchangeRateUtil;
import com.foreveross.fetl.queue.util.HangxinDateUtil;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.FetlErrorUtils;
import com.foreveross.fetl.util.HbaseToPlaneInfoUtil;
import com.foreveross.fetl.util.JsonUtil;
import com.foreveross.fetl.util.PlaneInfoUtil;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;
import com.foreveross.fetl.util.ZipUtils;

public class FormHbaseThread extends Thread{
	private static Logger log = Logger.getLogger("FormHbaseThread");
	private static List<Thread> runningThreads = new ArrayList<Thread>(); 
	private StringBuffer results;
	private StringBuffer keys;
	private ZipOutputStream hxZIP;
	private ZipOutputStream localZIP;
	private String source;
	private List<String> lines;
	private  Date time;
	private Date endTime;
	
	public FormHbaseThread(StringBuffer results,StringBuffer keys,ZipOutputStream hxZIP,ZipOutputStream localZIP,String source, List<String> lines, Date time,Date endTime){
		this.results=results;
		this.keys=keys;
		this.hxZIP=hxZIP;
		this.localZIP=localZIP;
		this.source=source;
		this.lines=lines;
		this.time=time;
		this.endTime=endTime;
	}
	public void run() {
		regist(this);//线程开始时注册 
		findFormHbase(results,keys,hxZIP,localZIP,source,lines,time,endTime);
//		threadSignal.countDown();//计时器减1  
		unRegist(this);//线程结束时取消注册  
	}
	public void regist(Thread t){  
	    synchronized(runningThreads){   
	        runningThreads.add(t);  
	    }  
	} 
	public void unRegist(Thread t){  
	    synchronized(runningThreads){   
	        runningThreads.remove(t);  
	    }  
	}  
	public static boolean hasThreadRunning() {  
		return (runningThreads.size() > 0);//通过判断runningThreads是否为空就能知道是否还有线程未执行完  
		} 
	/**
	 * 从hbase里读数据，并生成txt文件
	 * @param source
	 * @param lines
	 * @param time
	 */
	private void findFormHbase(StringBuffer results,StringBuffer keys,ZipOutputStream hxZIP,ZipOutputStream localZIP,String source, List<String> lines, Date time,Date endTime) {
		int thousand=0;
		String path=ReportProperties.getFilePath()+"company/newFile/"+source+"/";
		Calendar cStart = Calendar.getInstance();
		cStart.setTime(time);
		Calendar cEnd = Calendar.getInstance();
		cEnd.setTime(endTime);
		int i=1;
		int count=0;
		String id=null;
		keys.append("渠道： "+ source);
		StringBuilder josnString=new StringBuilder(512*1024);
		String  separator=System.getProperty("line.separator");
		List<String> listkeys=new ArrayList<String>();
		Set<Date> flightDateSet=HangxinDateUtil.getFlightDateSet();
		Integer[] returnDays=HangxinDateUtil.getReturnDays();
		int flag=0;
		if(lines.size()>0){
			while (cStart.before(cEnd)) {
				String timeStr=DateUtils.format(cStart.getTime(), "yyyyMMddHH");
				for(String line:lines){
					String[] citys=line.split("-");
					for(Date flightDate:flightDateSet){
						String goDateStr=DateUtils.format(flightDate, "yyyyMMdd");
						for(int j=0;j<returnDays.length;j++){
							ResultScanner rs = null;
							Calendar c =Calendar.getInstance();
							c.setTime(flightDate);
							c.add(Calendar.DAY_OF_MONTH, returnDays[j]);
							Date returnFlightDate=c.getTime();
							String returnDay=DateUtils.format(returnFlightDate, "yyyyMMdd");
							String startRow=RowkeyUtils.getRowKey(timeStr, source,citys[0],citys[1],goDateStr,returnDay, RowkeyUtils.START_ROW_KEY);
							String endRow=RowkeyUtils.getRowKey(timeStr, source, citys[0],citys[1],goDateStr,returnDay, RowkeyUtils.STOP_ROW_KEY);
							try {
								//查询Hbase
								try {
									rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
											startRow, endRow,new String[]{"planeinfo","transitsInfo","cabinsInfo","stopOversInfo",
											"returnPlaneInfosInfo","returnCabinsInfo","returnStopOversInfo","returnTransitsInfo"},null);
									rs.iterator();
								} catch (Exception e) {
									rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
											startRow, endRow,new String[]{"planeinfo","transitsInfo","cabinsInfo","stopOversInfo",
											"returnPlaneInfosInfo","returnCabinsInfo","returnStopOversInfo","returnTransitsInfo"},null);
									//如果从hbase里读数据失败了就再来一次
								}
								if(rs !=null){
								for (Result res : rs) {
									try {
										Map<String, Map<String, String>> curMap = HBaseHelper
												.transformHBaseRowToMap(res);
										Map<String, String> curPalneInfoMap = curMap.get("planeinfo");
										
										id =new String(res.getRow());
										listkeys.add(id);
										if(i==1){
											keys.append(" 开始rowkey： "+ id+"<br>");
											i++;
										}
										DoublePlaneInfo planeInfo= null;
										double refePrice=ExchangeRateUtil.getRefePrice(curPalneInfoMap.get("currency"));
										if(curPalneInfoMap.get("grabChannelNum").equals("00006") && null==curPalneInfoMap.get("currency")){
											refePrice=ExchangeRateUtil.getRefePrice("USD");
										}
										planeInfo=HbaseToPlaneInfoUtil.hbaseToPlaneInfo(curMap, refePrice,DoublePlaneInfo.class);
										if(null!=planeInfo && null!=planeInfo.getFlightNo()){
											List<PlaneInfo> list2=PlaneInfoUtil.changeToPlaneInfo(planeInfo);
											if(null!=list2 && list2.size()>0){
												for(PlaneInfo vo:list2){
													josnString.append(JsonUtil.getJSONString(vo));
													josnString.append(separator);
													count++;
												}
											}
										}
										if(count>=10000){//存在数据，则生成txt文件
											thousand++;
											count=0;
											flag++;
											String fileName=source+"_"+DateUtils.format(endTime, "yyyy-MM-dd_HH")+"_"+thousand+".txt";
											ZipUtils.creatFile(path, fileName, josnString.toString());
											josnString.delete(0, josnString.length());
											//生成的txt文件加载到.zip
											//加载到本地
											File file=new File(path+"/"+fileName);
											intoZIP(file, localZIP);
											intoZIP(file, hxZIP);
											new File(path+"/"+fileName).delete();
											listkeys.clear();
											count=0;
										}
											
									} catch (Exception e) {
										FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
												this.getClass().getName(),DateUtils.format(endTime, "yyyyMMddHH"), e);
										vo.setKeys(listkeys);
										FetlErrorUtils.saveToHbase(vo);
										listkeys.clear();
										count=0;
									}
								}
							}
							} catch (Exception e) {
								log.error(StringUtils.getStackTrace(e));
								FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
										this.getClass().getName(),DateUtils.format(endTime, "yyyyMMddHH"), e);
								vo.setStartRowKey(startRow);
								vo.setEndRowkey(endRow);
								FetlErrorUtils.saveToHbase(vo);
							}finally{
								if(rs != null) rs.close();
								rs = null;
							}
						}
					}
				}
				cStart.add(Calendar.HOUR_OF_DAY, 1);// 开始时间加一小时
			}
		}
		try {
			String fileName =null;
			if(count>0){
			//无论有无数据都创建路径
				thousand++;
				fileName=source+"_"+DateUtils.format(endTime, "yyyy-MM-dd_HH")+"_"+thousand+".txt";
				ZipUtils.creatFile(path, fileName, josnString.toString());
				josnString.delete(0, josnString.length());
				File file=new File(path+"/"+fileName);
				intoZIP(file, localZIP);
				intoZIP(file, hxZIP);
				new File(path+"/"+fileName).delete();
				listkeys.clear();
			}
			if(lines.size()==0){
				thousand++;
				fileName=source+"_"+DateUtils.format(endTime, "yyyy-MM-dd_HH")+"_"+thousand+".txt";
				ZipUtils.creatFile(path, fileName, josnString.toString());
				josnString.delete(0, josnString.length());
				File file=new File(path+"/"+fileName);
				intoZIP(file, localZIP);
				intoZIP(file, hxZIP);
				new File(path+"/"+fileName).delete();
				listkeys.clear();
			}
			if(flag==0 && lines.size()>0){
				thousand++;
				fileName=source+"_"+DateUtils.format(endTime, "yyyy-MM-dd_HH")+"_"+thousand+".txt";
				ZipUtils.creatFile(path, fileName, josnString.toString());
				josnString.delete(0, josnString.length());
				File file=new File(path);
				intoZIP(file, localZIP);
				intoZIP(file, hxZIP);
				new File(path+"/"+fileName).delete();
				listkeys.clear();
			}
				
			
		} catch (Exception e) {
			FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
					this.getClass().getName(),DateUtils.format(endTime, "yyyyMMddHH"), e);
			vo.setKeys(listkeys);
			FetlErrorUtils.saveToHbase(vo);
			listkeys.clear();
		}
		if(i>1){
			keys.append(" 结束rowkey： "+ id +"<br>");	
		}else{
			keys.append(" 没数据 <br> ");
		}
		
	}

	/**
	 * 加载文件至.ZIP（压缩至.zip）
	 * @param file 需要加载的文件
	 * @param zipFile zip包
	 * @throws Exception 
	 */
	public void intoZIP(File file,ZipOutputStream zipFile){
		try {
			log.error("txt文件加载至.zip文件");
			ZipUtils.zipFile(file, zipFile);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}
